package com.spzx.common.rabbit.config;

import com.alibaba.fastjson2.JSON;
import com.spzx.common.rabbit.entity.GuiguCorrelationData;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.concurrent.TimeUnit;

@Slf4j
public class RabbitInitConfigApplicationListener implements ApplicationListener<ApplicationReadyEvent> {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private RedisTemplate redisTemplate;


    /**
     * springBootApplication 准备事件触发
     * <p>
     * 绑定生产者的消息确认和失败消息退回
     */
    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        this.setupCallbacks();
    }

    private void setupCallbacks() {
        /**
         * 只确认消息是否正确到达 Exchange 中,成功与否都会回调
         *
         * @param correlation 相关数据  非消息本身业务数据
         * @param ack         应答结果
         * @param reason      如果发送消息到交换器失败，错误原因
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, reason) -> {
            if (ack) {
                log.info("消息发送到Exchange成功：{}", correlationData);
            } else {
                //消息到交换器失败
                log.error("消息发送到Exchange失败：{}", reason);
                //重试
                this.retrySendMsg(correlationData);
            }
        });

        /**
         * RabbitMQ的生产者消息退回机制用于处理消息成功发送到交换机但无法路由到任何队列的情况
         * 消息没有正确到达队列时触发回调，如果正确到达队列则不执行
         *
         * 默认情况下mandatory的值是false
         * 但是只要配置开启了publisher-returns则mandatory的值就是true
         * this.rabbitTemplate.setMandatory(true); // 是否让rabbitmq将失败的消息的信息再次返回给生产者
         */
        rabbitTemplate.setReturnsCallback((returned) -> {
            log.error("返回: " + returned.getMessage()
                    + "\n 响应码: " + returned.getReplyCode()
                    + "\n 响应消息: " + returned.getReplyText()
                    + "\n 交换机: " + returned.getExchange()
                    + "\n 路由: " + returned.getRoutingKey());

            //当路由队列失败 需要重发
            //1.构建相关数据对象
            String redisKey = returned.getMessage().getMessageProperties().getHeader("spring_returned_message_correlation");
            String correlationDataStr = (String) redisTemplate.opsForValue().get(redisKey);
            GuiguCorrelationData guiguCorrelationData = JSON.parseObject(correlationDataStr, GuiguCorrelationData.class);
            //2.调用消息重发方法
            this.retrySendMsg(guiguCorrelationData);
        });
    }

    @SneakyThrows
    private void retrySendMsg(CorrelationData correlationData) {
        //获取相关数据
        GuiguCorrelationData guiguCorrelationData = (GuiguCorrelationData) correlationData;

        //获取redis中存放的重试次数
        //先将次数写回到redis中，再重发
        int retryCount = guiguCorrelationData.getRetryCount();
        if (retryCount >= 3) {
            //超过最大重试次数
            log.error("生产者超过最大重试次数，将失败的消息存入数据库用人工处理；给管理员发送邮件；给管理员发送短信；");
            return;
        }

        Thread.sleep(1000);

        //重发次数+1
        retryCount += 1;
        guiguCorrelationData.setRetryCount(retryCount);

        //将相关数据存入redis
        redisTemplate.opsForValue().set(
                guiguCorrelationData.getId(),
                JSON.toJSONString(guiguCorrelationData),
                10,
                TimeUnit.MINUTES);

        //重发消息
        log.info("进行消息重发！");
        rabbitTemplate.convertAndSend(
                guiguCorrelationData.getExchange(),
                guiguCorrelationData.getRoutingKey(),
                guiguCorrelationData.getMessage(),
                guiguCorrelationData
        );
    }
}
